/**
 *    Copyright (c) [2022] [中和农信]
 *    rabbit-mq-plus is licensed under Mulan PSL v2.
 *    You can use this software according to the terms and conditions of the Mulan PSL v2.
 *    You may obtain a copy of Mulan PSL v2 at:
 *             http://license.coscl.org.cn/MulanPSL2
 *    THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 *    See the Mulan PSL v2 for more details.
 *
 */
package com.cfpamf.athean.rabbit.mq.plus.consumer.core.impl;

import java.time.LocalDateTime;

import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import com.cfpamf.athean.rabbit.mq.plus.common.constant.BusinessConstants;
import com.cfpamf.athean.rabbit.mq.plus.common.exception.RabbitMqPlusException;
import com.cfpamf.athean.rabbit.mq.plus.domain.builder.RabbitMqPlusConsumerTaskBuilder;
import com.cfpamf.athean.rabbit.mq.plus.domain.dto.RabbitMqPlusConsumerDTO;
import com.cfpamf.athean.rabbit.mq.plus.domain.po.RabbitMqPlusConsumerTaskPO;
import com.cfpamf.athean.rabbit.mq.plus.service.RabbitMqPlusConsumerTaskService;

/**
 * 类DefaultRabbitMqPlusConsumerHandler.java的实现描述：
 *
 * @author Andy（陈洪桥）2022年12月12 21:24
 */
@Service("rabbitMqPlusConsumerHandler")
public class DefaultRabbitMqPlusConsumerHandler implements RabbitMqPlusConsumerHandler {

    @Override
    @Transactional
    public void consumerProcessor(RabbitMqPlusConsumerTaskService rabbitMqPlusConsumerTaskService,
        RabbitMqPlusConsumerDTO consumerDTO, String consumerTag, int retryCount, LocalDateTime firstConsumerTime,
        String messageId, String data) {
        try {
            // 首次处理，写DB记录
            this.firstConsumerSave(rabbitMqPlusConsumerTaskService, consumerDTO, retryCount, firstConsumerTime,
                messageId, data);
            // 业务消息消费处理
            boolean isAck = consumerDTO.getRabbitMqPlusConsumerListener().onMessage(messageId, data, consumerTag,
                BusinessConstants.ZERO);
            // 业务处理失败
            this.consumerProcessorFailure(rabbitMqPlusConsumerTaskService, consumerDTO, retryCount, firstConsumerTime,
                messageId, data, isAck);
            // 业务处理成功，更新DB数据状态
            this.consumerProcessorSuccess(rabbitMqPlusConsumerTaskService, consumerDTO, retryCount, isAck);
        } catch (Exception e) {
            throw new RabbitMqPlusException(
                String.format("[rabbitMqPlus]消费者，业务消息消费处理失败，异常，QueueName：%s", consumerDTO.getQueueName()), e);
        }
    }

    private void firstConsumerSave(RabbitMqPlusConsumerTaskService rabbitMqPlusConsumerTaskService,
        RabbitMqPlusConsumerDTO consumerDTO, int retryCount, LocalDateTime firstConsumerTime, String messageId,
        String data) {
        if (retryCount > BusinessConstants.ZERO || retryCount < BusinessConstants.ZERO) {
            return;
        }
        // 首次处理，写DB记录
        RabbitMqPlusConsumerTaskPO consumerTaskPO = RabbitMqPlusConsumerTaskBuilder
            .buildFirstConsumerTaskRecord(retryCount, consumerDTO, firstConsumerTime, messageId, data);
        rabbitMqPlusConsumerTaskService.save(consumerTaskPO);
    }

    private void consumerProcessorFailure(RabbitMqPlusConsumerTaskService rabbitMqPlusConsumerTaskService,
        RabbitMqPlusConsumerDTO consumerDTO, int retryCount, LocalDateTime firstConsumerTime, String messageId,
        String data, boolean isAck) {
        if (Boolean.TRUE.equals(isAck)) {
            return;
        }
        RabbitMqPlusConsumerTaskPO consumerTaskPO = RabbitMqPlusConsumerTaskBuilder
            .buildConsumerProcessorFailureTaskRecord(retryCount, consumerDTO, firstConsumerTime, data);
        // 更新DB消费记录处理结果
        rabbitMqPlusConsumerTaskService.updateByQueueNameAndConsumerTypeCode(consumerTaskPO);
    }

    private void consumerProcessorSuccess(RabbitMqPlusConsumerTaskService rabbitMqPlusConsumerTaskService,
        RabbitMqPlusConsumerDTO consumerDTO, int retryCount, boolean isAck) {
        if (Boolean.FALSE.equals(isAck)) {
            return;
        }
        RabbitMqPlusConsumerTaskPO consumerTaskPO =
            RabbitMqPlusConsumerTaskBuilder.buildConsumerProcessorSuccess(retryCount, consumerDTO);
        // 更新DB消费记录处理成功结果
        rabbitMqPlusConsumerTaskService.updateByQueueNameAndConsumerTypeCode(consumerTaskPO);
    }

}
